package net.bwie.realtime.jtp.dwd.log.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class DataSplitFuction extends ProcessFunction<String,String> {
    private final OutputTag<String> errTag;
    private final OutputTag<String> startTag;
    private final OutputTag<String> displayTag;
    private final OutputTag<String> actionsTag;
    public DataSplitFuction(OutputTag<String> errTag, OutputTag<String> startTag,
                            OutputTag<String> displayTag, OutputTag<String> actionsTag){
        this.errTag = errTag;
        this.startTag = startTag;
        this.displayTag = displayTag;
        this.actionsTag = actionsTag;
    }
    @Override
    public void processElement(String value,Context ctx, Collector<String> out) throws Exception {
        JSONObject jsonObject = JSON.parseObject(value);
        //错误日志
        if(jsonObject.get("err") != null){
            ctx.output(errTag,value);
            //已有错误日志,删除
            jsonObject.remove("err");
        }
        //启动日志
        if(jsonObject.get("start") != null){
            ctx.output(startTag,jsonObject.toJSONString());
        }
        //页面日志
        else {
            JSONArray displayArray = jsonObject.getJSONArray("displays");
            JSONArray actionsArray = jsonObject.getJSONArray("actions");
            //删除
            jsonObject.remove("displays");
            jsonObject.remove("actions");
            out.collect(jsonObject.toJSONString());
            //曝光日志
            if(displayArray != null && !displayArray.isEmpty()){
                for(int i = 0 ;i < displayArray.size();i++){
                    JSONObject display = displayArray.getJSONObject(i);
                    jsonObject.put("display",display);
                    ctx.output(displayTag,jsonObject.toJSONString());
                }
            }
            jsonObject.remove("display");
            //动作日志
            if(actionsArray != null && !actionsArray.isEmpty()){
                for(int i = 0;i < actionsArray.size();i++){
                    JSONObject actions = actionsArray.getJSONObject(i);
                    jsonObject.put("action",actions);
                    ctx.output(actionsTag,jsonObject.toJSONString());
                }
            }
        }
    }
}
